package com.awesome.spring;

import com.awesome.spring.model.UserSleepStatus;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.time.Instant;
import java.util.Date;
import java.util.Properties;
import java.util.Random;

/**
 * @program: u-InternetOfThings
 * @description: sleep状态流
 * @author: Alex Wu
 * @createDate: 2025-03-20 17:54
 **/

public class SleepStatusStreamJob {

    private static final String ENV_KAFKA_SERVER = "KAFKA_SERVER";
    private static final String KAFKA_BEHAVIOR_LOG_TOPIC = "test";
    private static final String KAFKA_BEHAVIOR_LOG_GROUP_ID = "test_group";
    private static final String JOB_NAME_BEHAVIOR_LOG = "Flink_Behavior_Log";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 Kafka 数据源（假设从 Kafka 获取用户状态数据）
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.102.8:9092");
        properties.setProperty("group.id", "flink-group");

        /*FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                "test",
                new SimpleStringSchema(),
                properties
        );*/

        // 读取 Kafka 数据流
//        DataStream<String> sourceStream = env.addSource(kafkaConsumer);


        DataStreamSource<UserSleepStatus> sourceStream = env.addSource(new SourceFunction<UserSleepStatus>() {
            private boolean flag = true;

            private Integer startTime = 1000;

            @Override
            public void run(SourceContext<UserSleepStatus> ctx) throws Exception {
                Random random = new Random();
                while (flag) {
                  /*  String orderId = UUID.randomUUID().toString();
                    int userId = random.nextInt(2);
                    int money = random.nextInt(101);
                    //随机模拟延迟时间在5秒内
                    long eventTime = System.currentTimeMillis() - random.nextInt(5) * 1000;*/

                    Long userId = 1L;
                    String status = "sleep";
                    startTime += 1000;
                    Long eventTime = Instant.parse("2024-11-15T00:00:00Z").toEpochMilli() - random.nextInt(5) * 1000 + startTime;
//                    Long eventTime = System.currentTimeMillis() - random.nextInt(5) * 1000;
                    Integer breathingRate =  random.nextInt(100) ;
                    Integer heartRate =  random.nextInt(100) ;
                    Integer turnsNum =  random.nextInt(100) ;
                    UserSleepStatus userSleepStatus = new UserSleepStatus(userId, status, eventTime, breathingRate, heartRate, turnsNum);



//                    ctx.collect(JSON.toJSONString(userSleepStatus));
                    ctx.collect(userSleepStatus);


                    long currentTimeMillis = System.currentTimeMillis();

                    // 创建SimpleDateFormat对象，指定格式
                    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");

                    // 将时间戳转换为Date对象
                    Date date = new Date(currentTimeMillis);

                    // 格式化Date对象，得到字符串形式的时间
                    String formattedDate = sdf.format(date);

//                    System.out.println("Formatted time: " + formattedDate + " event time: " + sdf.format(userSleepStatus.getEventTime()) + " Order info: " + userSleepStatus );


                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        });



        SingleOutputStreamOperator<UserSleepStatus> userSleepStatusWatermark = sourceStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<UserSleepStatus>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 允许延迟 30 秒
                        .withTimestampAssigner((event, timestamp) -> event.getEventTime())
        ).map(new MapFunction<UserSleepStatus, UserSleepStatus>() {
            @Override
            public UserSleepStatus map(UserSleepStatus userSleepStatus) {
//                UserSleepStatus userSleepStatus = JSON.parseObject(value, UserSleepStatus.class);
                long currentTimeMillis = System.currentTimeMillis();

                // 创建SimpleDateFormat对象，指定格式
                SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");

                // 将时间戳转换为Date对象
                Date date = new Date(currentTimeMillis);

                // 格式化Date对象，得到字符串形式的时间
                String formattedDate = sdf.format(date);

                System.out.println("Formatted time: " + formattedDate + " event time: " + sdf.format(userSleepStatus.getEventTime())  + "收到数据--> UserSleepStatus: " +  userSleepStatus);
                return userSleepStatus;
            }
        });// **绑定水位线***/

        SingleOutputStreamOperator<String> process = userSleepStatusWatermark
                .keyBy(UserSleepStatus::getUserId)  // RickProcessFunc 中使用的 Key
                .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 1分钟窗口
                .process(new ProcessWindowFunction<UserSleepStatus, String, Long, TimeWindow>() {

                    @Override
                    public void process(Long key, Context context, Iterable<UserSleepStatus> elements, Collector<String> collector) throws Exception {


                        for (UserSleepStatus userSleepStatus : elements) {
                            System.out.println("In Windows -->" + userSleepStatus);    // 打印此窗口内的数据
                        }

                        long windowStart = context.window().getStart();
                        long windowEnd = context.window().getEnd();
                        // Format window timestamp
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String formattedWindowStart = sdf.format(new Date(windowStart));
                        String formattedWindowEnd = sdf.format(new Date(windowEnd));
                        String curTime = sdf.format(new Date());

                        String format = String.format("CurTime: %s, Window [%s - %s], UserId: %d",
                                curTime, formattedWindowStart, formattedWindowEnd, key);
                        System.out.println(format);

                        collector.collect("============");
                    }
                });

        process.print();


       /* // 每1分钟聚合，计算最大值和最小值
        DataStream<UserStatusAggregated> aggregatedStream = userSleepStatusWatermark
                .keyBy(UserSleepStatus::getUserId)
                .window(TumblingEventTimeWindows.of(Time.minutes(1))) // 1分钟窗口
                .aggregate(new MaxMinAggregation());

        // 30分钟周期计算
        DataStream<UserStatusAggregated> finalAggregatedStream = aggregatedStream
                .window(TumblingEventTimeWindows.of(Time.minutes(30))) // 30分钟汇总窗口
                .aggregate(new FinalAggregation()); // 计算每30分钟的聚合状态

        // 存储到MySQL
        finalAggregatedStream.addSink(new MySQLSink());*/

        env.execute("User Status Calculation");
    }
}
